package com.xiehua.task.job;

import com.xiehua.task.actor.cmd.NotifyClient;
import com.xiehua.task.job.bucket.CronBucket;
import com.xiehua.task.job.bucket.DelayBucket;
import com.xiehua.task.job.bucket.FixedBucket;
import com.xiehua.task.job.bucket.TopicBucket;
import com.xiehua.task.job.cron.CronSequenceGenerator;
import com.xiehua.task.job.pool.DelayJob;
import com.xiehua.task.job.pool.FixedJob;
import com.xiehua.task.job.pool.Job;
import com.xiehua.task.persistence.TaskPersistence;
import com.xiehua.task.persistence.rocks.RocksPersitence;
import com.xiehua.task.persistence.rocks.RocksPool;
import com.xiehua.task.persistence.rocks.XeihuaPersitence;
import org.apache.commons.pool2.impl.GenericObjectPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tio.core.ChannelContext;
import org.tio.core.GroupContext;
import org.tio.utils.json.Json;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Transaction;

import java.io.IOException;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZoneOffset;
import java.util.Date;
import java.util.List;
import java.util.stream.Collectors;

import static com.xiehua.task.job.MachiningCenter.JOB_BUCKET_PIX;
import static com.xiehua.task.job.bucket.ReadyBucket.JOB_READY_QUEUE;
import static com.xiehua.task.job.pool.DelayJobPool.JOB_POOL_DELAY_KEY;
import static com.xiehua.task.job.pool.FixedJobPool.FIXED_JOB_KEY_PIX;
import static com.xiehua.task.job.pool.FixedJobPool.JOB_POOL_FIXED_KEY;
import static com.xiehua.task.job.pool.Job.JobStatus.notify;
import static com.xiehua.task.job.pool.Job.JobType.delay;
import static com.xiehua.task.job.pool.Job.JobType.fixed;
import static com.xiehua.task.prop.XiehuaConfig.REVISE_TIME;

public class TaskReverseDefualt extends RidesOperationDefualt implements TaskReverse {

    private static Logger log = LoggerFactory.getLogger(TaskReverseDefualt.class);

    public TaskReverseDefualt(Jedis jedis) {
        super(jedis);
    }


    /**
     * //不可执行状态，等待时钟周->job初始化
     *
     * @param job
     */
    @Override
    public Job delay(Job job, GroupContext context) {
        if (job == null) return null;
        job.setStatus(Job.JobStatus.delay);
        if (delay.equals(job.getType())) {
            DelayJob delayJob = (DelayJob) job;
            return new DelayBucket(jedis).addJob(delayJob);
        }
        if (fixed.equals(job.getType())) {
            FixedJob fixedJob = (FixedJob) job;
            return regFixedJob(fixedJob, context);
        }
        log.error("任务为空或者类型不支持:{}", job.toString());
        return null;
    }

    /***
     * 注册定时任务
     * **/
    private FixedJob regFixedJob(FixedJob fixedJob, GroupContext context) {
        //check reg fixed task whether or not
        String key = FIXED_JOB_KEY_PIX + fixedJob.getTopic()+"_"+ fixedJob.getClassName() + "_" + fixedJob.getMethodName();
        if(jedis.exists(key)) return new FixedBucket(jedis).queryJob(jedis.get(key));
        //put topic bucket
        TopicBucket topicBucket = new TopicBucket(jedis);
        List<NotifyClient> list = topicBucket.querySurvivalChannelByTopicName(fixedJob.getTopic(), context);
        if (list == null || list.size() < 1) return null;
        //add cron
        CronBucket cronBucket = new CronBucket(jedis);
        cronBucket.addCron(fixedJob.getTopic(), fixedJob.getClassName(), fixedJob.getMethodName(), fixedJob.getCron());
        //add fixed job
        CronSequenceGenerator cronSequenceGenerator = new CronSequenceGenerator(fixedJob.getCron());
        LocalDateTime now = LocalDateTime.now();
        Date next = cronSequenceGenerator.next(Date.from(now.atZone(ZoneId.systemDefault()).toInstant()));
        Long delay = next.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime().toEpochSecond(ZoneOffset.of("+8")) - now.toEpochSecond(ZoneOffset.of("+8"));
        fixedJob.setDelay(delay);
        fixedJob = new FixedBucket(jedis).addJob(fixedJob);
        //add fixed job flag
        jedis.setex(key,delay.intValue() - REVISE_TIME,fixedJob.getJobId());
        return fixedJob;
    }

    /***
     * 可执行状态，等待消费。
     *
     * @param jobs
     **/
    @Override
    public void ready(List<Job> jobs) {
        Transaction transaction = jedis.multi();
        try {
            List<String> serializJob = jobs.stream().map(s -> {
                s.setStatus(Job.JobStatus.ready);
                s.setUpdateTime(LocalDateTime.now());
                String t = Json.toJson(s);
                if (delay.equals(s.getType())) transaction.hset(JOB_POOL_DELAY_KEY, s.getJobId(), t);//更新job pool
                if (fixed.equals(s.getType())) transaction.hset(JOB_POOL_FIXED_KEY, s.getJobId(), t);//更新job pool
                transaction.zrem(JOB_BUCKET_PIX + s.getTopic(), s.getJobId());//从bucket移除
                return t;
            }).collect(Collectors.toList());
            transaction.lpush(JOB_READY_QUEUE, serializJob.toArray(new String[serializJob.size()]));//job放入queue
            transaction.exec();
        } catch (Exception e) {
            transaction.discard();
        } finally {
            try {
                transaction.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 任务已通知，但还未得到消费者的响应
     *
     * @param job
     **/
    @Override
    public void notify(Job job,GroupContext context) {
        if (job == null) return;
        job.setStatus(notify);//任务已通知，但还未得到消费者的响应
        job.setUpdateTime(LocalDateTime.now());
        if (delay.equals(job.getType())) {
            DelayJob delayJob = (DelayJob) job;
            new DelayBucket(jedis).updateJob(delayJob);
        }
        if (fixed.equals(job.getType())) {
            FixedJob fixedJob = (FixedJob) job;
            new FixedBucket(jedis).updateJob(fixedJob);
            //重新注册定时任务
            FixedJob reFixedJob = new FixedJob(fixedJob.getTopic(), fixedJob.getClassName(), fixedJob.getMethodName(), fixedJob.getCron());
            delay(reFixedJob, context);
        }
    }

    /**
     * 客户端响应完成
     *
     * @param jobId
     **/
    @Override
    public void executeed(String jobId, ChannelContext channelContext) {
        Job job = queryJob(jobId);
        if (job == null) return;
        job.setStatus(Job.JobStatus.executeed);
        job.setUpdateTime(LocalDateTime.now());
        if (delay.equals(job.getType())) {
            DelayJob delayJob = (DelayJob) job;
            new DelayBucket(jedis).updateJob(delayJob);
        }
        if (fixed.equals(job.getType())) {
            FixedJob fixedJob = (FixedJob) job;
            new FixedBucket(jedis).updateJob(fixedJob);
        }
    }

    /**
     * 任务执行成功
     **/
    @Override
    public void success(String jobId) throws Exception{
        Job job = queryJob(jobId);
        if (job == null) return;
        job.setStatus(Job.JobStatus.success);
        job.setUpdateTime(LocalDateTime.now());
        if (delay.equals(job.getType())) {
            DelayJob delayJob = (DelayJob) job;
            new DelayBucket(jedis).updateJob(delayJob);
        }
        if (fixed.equals(job.getType())) {
            FixedJob fixedJob = (FixedJob) job;
            new FixedBucket(jedis).updateJob(fixedJob);
        }
        //persitence
        persitenceJob(job);
    }

    /**
     * 执行失败
     **/
    @Override
    public void fail(String jobId) throws Exception{
        Job job = queryJob(jobId);
        if (job == null) return;
        job.setStatus(Job.JobStatus.fail);
        job.setUpdateTime(LocalDateTime.now());
        if (delay.equals(job.getType())) {
            DelayJob delayJob = (DelayJob) job;
            new DelayBucket(jedis).updateJob(delayJob);
        }
        if (fixed.equals(job.getType())) {
            FixedJob fixedJob = (FixedJob) job;
            new FixedBucket(jedis).updateJob(fixedJob);
        }
        //persitence
        persitenceJob(job);
    }

    /**
     * 根据jobId查询job
     **/
    private Job queryJob(String jobId) {
        DelayJob delayJob = new DelayBucket(jedis).queryJob(jobId);
        if (delayJob != null) return delayJob;
        FixedJob fixedJob = new FixedBucket(jedis).queryJob(jobId);
        if (fixedJob != null) return fixedJob;
        return null;
    }

    /**
     * 任务持久化
     **/
    private void persitenceJob(Job job) throws Exception{
        GenericObjectPool<XeihuaPersitence> pool = RocksPool.getInstance();
        XeihuaPersitence persitence = pool.borrowObject();
        persitence.persistence(job);
        pool.returnObject(persitence);
    }
}
